package cn.iocoder.yudao.module.vos.job;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
import cn.iocoder.yudao.framework.quartz.core.handler.JobHandler;
import cn.iocoder.yudao.module.vos.dal.dataobject.activityfailednumbers.ActivityFailedNumbersDO;
import cn.iocoder.yudao.module.vos.dal.dataobject.activitytask.ActivityTaskDO;
import cn.iocoder.yudao.module.vos.dal.dataobject.marketinglistlog.MarketingListLogDO;
import cn.iocoder.yudao.module.vos.dal.dataobject.marketingscore.MarketingScoreDO;
import cn.iocoder.yudao.module.vos.dal.dataobject.pushlistconfig.PushListConfigDO;
import cn.iocoder.yudao.module.vos.dal.mysql.activityfailednumbers.ActivityFailedNumbersMapper;
import cn.iocoder.yudao.module.vos.dal.mysql.activitytask.ActivityTaskMapper;
import cn.iocoder.yudao.module.vos.dal.mysql.marketingscore.MarketingScoreMapper;
import cn.iocoder.yudao.module.vos.dal.mysql.pushlistconfig.PushListConfigMapper;
import cn.iocoder.yudao.module.vos.job.vo.PushItem;
import cn.iocoder.yudao.module.vos.service.marketinglistlog.MarketingListLogService;
import cn.iocoder.yudao.module.vos.utils.CdrUtil;
import cn.iocoder.yudao.module.vos.utils.InterfaceAClient;
import cn.iocoder.yudao.module.vos.utils.TideTemplateSmsClient;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;

/**
 * 拉取昨日日表 TopN，按配置分组发送；
 * 仅当对应营销工具发送成功时，才落库 vos_marketing_list_log。
 *
 * 新逻辑：
 *  - 先按 tenantId 分组；
 *  - 对每个 tenantId 再按 marketingTools 分组并逐工具发送；
 *  - 一个 tenantId 下所有营销工具执行完成后，
 *    将该租户中“发送成功”的 PushItem 去重后统一 persistGroupLogs。
 */
@Component
@Slf4j
public class PushListDispatchJob implements JobHandler {

    @Resource
    private PushListConfigMapper pushListConfigMapper;
    @Resource
    private MarketingScoreMapper marketingScoreMapper;
    @Resource
    private MarketingListLogService marketingListLogService;

    @Resource
    private InterfaceAClient interfaceAClient;

    @Resource
    private ActivityTaskMapper activityTaskMapper;

    @Resource
    private ActivityFailedNumbersMapper activityFailedNumbersMapper;

    @Override
    public String execute(String param) {
        Map<String, String> p = Query.parse(param);
        String day = p.get("day");
        String day8 = (day == null || day.isEmpty())
                ? LocalDate.now().minusDays(1).format(DateTimeFormatter.ofPattern("yyyyMMdd"))
                : toDay8(day);
        String listDate = toYmd(day8); // yyyy-MM-dd

        String table = "submeter_marketing_score_" + day8;
        log.info("[push-list] start. table={}, day8={}, listDate={}", table, day8, listDate);

        // 1) 读取启用的推送配置
        List<PushListConfigDO> configs = pushListConfigMapper.selectList(
                new LambdaQueryWrapperX<PushListConfigDO>().eq(PushListConfigDO::getEnabled, true));
        if (configs == null || configs.isEmpty()) {
            log.info("[push-list] no enabled configs.");
            return "no enabled configs";
        }

        // 2) 汇总 items
        int rawPicked = 0;
        int cfgBatches = 0;
        List<PushItem> items = new ArrayList<>();

        for (PushListConfigDO cfg : configs) {
            String lvl = cfg.getListLevel();
            Integer limit = cfg.getPullCount();
            Long tenantId = cfg.getRechargeTenantId();
            String marketingTools = cfg.getMarketingTools();

            if (StrUtil.isBlank(lvl) || limit == null || limit <= 0) {
                log.warn("[push-list] skip cfg id={} level={} limit={}", cfg.getId(), lvl, limit);
                continue;
            }

            Set<String> alreadyConvertedPhones = getConvertedPhones(tenantId);

            List<MarketingScoreDO> rows;
            try {
                rows = marketingScoreMapper.selectTopByLevelFromTable(table, lvl, limit);
            } catch (Exception e) {
                log.error("[push-list] read table fail. table={}, level={}, limit={}, err={}",
                        table, lvl, limit, e.getMessage(), e);
                continue;
            }
            if (rows == null || rows.isEmpty()) {
                log.info("[push-list] table={} level={} got 0 rows.", table, lvl);
                continue;
            }

            rawPicked += rows.size();
            cfgBatches++;

            for (MarketingScoreDO r : rows) {
                if (!alreadyConvertedPhones.contains(r.getPhoneE164())) {
                    PushItem item = toItem(tenantId, listDate, lvl, r, marketingTools);
                    items.add(item);
                }
            }
        }

        int picked = items.size();
        log.info("[push-list] collected: rawPicked={}, uniquePicked={}, cfgBatches={}",
                rawPicked, picked, cfgBatches);
        if (picked == 0) {
            return String.format("batches=%d, picked=%d, day=%s", cfgBatches, 0, day8);
        }

        // 3) 先按 tenantId 分组，再在租户内部按工具分组发送；
        //    一个 tenantId 下所有工具调用完毕后，再统一落库（去重）
        simulateBatchSendAll(items);

        String msg = String.format("cfg_batches=%d, raw_picked=%d, unique_picked=%d, day=%s",
                cfgBatches, rawPicked, picked, day8);
        log.info("[push-list] done. {}", msg);
        return msg;
    }

    /**
     * score 行 -> 中间对象
     */
    private static PushItem toItem(Long tenantId, String listDate, String lvl,
                                   MarketingScoreDO r, String marketingTools) {
        PushItem x = new PushItem();
        x.setTenantId(tenantId);
        x.setListDate(listDate);
        x.setListLevel(lvl);
        x.setPhoneE164(r.getPhoneE164());
        x.setPhoneMd5(r.getPhoneMd5());
        x.setScore(r.getScore());
        x.setPhoneTags(r.getPhoneTags());
        x.setMarketingTools(marketingTools);
        return x;
    }

    /**
     * 查询已成交手机号
     */
    private Set<String> getConvertedPhones(Long tenantId) {
        List<MarketingListLogDO> logs = marketingListLogService.getConvertedLogs(tenantId, 2);
        Set<String> phones = new HashSet<>();
        for (MarketingListLogDO log : logs) {
            phones.add(log.getPhoneE164());
        }
        return phones;
    }

    /**
     * 分组发送（新版逻辑）：
     * 1）先按 tenantId 分组；
     * 2）每个 tenantId 内部再按 marketingTools 分组（可能有多个工具，用逗号分隔）；
     * 3）按工具调用外部接口，记录成功的 PushItem；
     * 4）一个 tenantId 执行完所有营销工具后，将该租户内所有成功记录去重后统一落库。
     */
    private void simulateBatchSendAll(List<PushItem> items) {
        log.info("[push-list] simulate ALL batch send => size={}", items.size());
        if (items == null || items.isEmpty()) {
            return;
        }

        // 1) 按 tenantId 分组
        Map<Long, List<PushItem>> groupedByTenant = new LinkedHashMap<>();
        for (PushItem item : items) {
            Long tenantId = item.getTenantId();
            groupedByTenant.computeIfAbsent(tenantId, k -> new ArrayList<>()).add(item);
        }

        // 2) 对每个 tenantId 处理
        for (Map.Entry<Long, List<PushItem>> tenantEntry : groupedByTenant.entrySet()) {
            Long tenantId = tenantEntry.getKey();
            List<PushItem> tenantItems = tenantEntry.getValue();
            log.info("[push-list] tenantId={} has {} items", tenantId, tenantItems.size());

            // 每个租户内：记录成功发送的名单，使用 successKeySet 去重
            Set<String> successKeySet = new HashSet<>();
            List<PushItem> successItems = new ArrayList<>();

            // 2.1) 在该租户内部，按营销工具 key 再分组
            Map<String, List<PushItem>> groupedByTool = new LinkedHashMap<>();
            for (PushItem item : tenantItems) {
                String toolsStr = StrUtil.blankToDefault(item.getMarketingTools(), "");
                if (toolsStr.isEmpty()) {
                    groupedByTool.computeIfAbsent("default", k -> new ArrayList<>()).add(item);
                    continue;
                }
                for (String toolKey : toolsStr.split(",")) {
                    String key = toolKey == null ? "" : toolKey.trim();
                    if (key.isEmpty()) {
                        continue;
                    }
                    groupedByTool.computeIfAbsent(key, k -> new ArrayList<>()).add(item);
                }
            }

            // 2.2) 对该 tenant 的每一个工具分组调用外部接口
            for (Map.Entry<String, List<PushItem>> entry : groupedByTool.entrySet()) {
                String key = entry.getKey(); // e.g. "94waihu-xxx", "TideMsg-123"
                List<PushItem> groupItems = entry.getValue();

                String[] parts = key == null ? new String[0] : key.split("-", -1);
                String tool = CdrUtil.getOrNull(parts, 0);
                String taskId = CdrUtil.getOrNull(parts, 1); // 对 94waihu/TideMsg/TideAi 场景

                log.info("[push-list] tenantId={} group by toolKey='{}', tool='{}', size={}",
                        tenantId, key, tool, groupItems.size());

                if ("94waihu".equalsIgnoreCase(tool)) {
                    // 94 外呼：按 10,000 条为上限拆分多批发送
                    final int MAX_BATCH = 10_000;
                    int total = groupItems.size();
                    int batchNo = 0;
                    for (int from = 0; from < total; from += MAX_BATCH) {
                        int to = Math.min(from + MAX_BATCH, total);
                        batchNo++;
                        List<PushItem> batch = new ArrayList<>(groupItems.subList(from, to));
                        log.info("[push-list] tenantId={} tool='{}' -> channel(taskId)='{}', batch #{}, range=[{}, {}), size={}",
                                tenantId, tool, taskId, batchNo, from, to, batch.size());

                        boolean sentOk = sendToInterface94(key, batch, taskId);
                        if (sentOk) {
                            // 该批发送成功，批次内记录到成功集合（去重）
                            for (PushItem it : batch) {
                                String sk = buildLogKey(it);
                                if (successKeySet.add(sk)) {
                                    successItems.add(it);
                                }
                            }
                        } else {
                            log.warn("[push-list] tenantId={} tool='{}' channel='{}' batch #{} send failed. skip logging for this batch.",
                                    tenantId, tool, taskId, batchNo);
                        }
                    }
                } else if ("TideMsg".equalsIgnoreCase(tool)) {
                    // Tide 模板短信：key 约定为 "TideMsg-{templateId}"，这里将 taskId 当作 templateId
                    if (StrUtil.isBlank(taskId)) {
                        log.warn("[push-list] tenantId={} tool='{}' templateId(taskId) is blank, skip this group. key={}, size={}",
                                tenantId, tool, key, groupItems.size());
                        continue;
                    }
                    Long templateId;
                    try {
                        templateId = Long.parseLong(taskId);
                    } catch (NumberFormatException e) {
                        log.error("[push-list] tenantId={} tool='{}' invalid templateId(taskId)='{}', key={}, skip this group.",
                                tenantId, tool, taskId, key);
                        continue;
                    }

                    log.info("[push-list] tenantId={} tool='{}' use templateId={} send to TIDE, size={}",
                            tenantId, tool, templateId, groupItems.size());

                    try {
                        // 调用 TIDE 批量模板短信：只从 PushItem 取 phoneE164
                        TideTemplateSmsClient.sendTemplateSmsForPushItems(groupItems, templateId);
                        // 调用成功即视为成功（如需严格校验 rtCode，可扩展为返回 boolean）
                        for (PushItem it : groupItems) {
                            String sk = buildLogKey(it);
                            if (successKeySet.add(sk)) {
                                successItems.add(it);
                            }
                        }
                    } catch (Exception e) {
                        log.error("[push-list] tenantId={} tool='{}' send to TIDE error. templateId={}, err={}",
                                tenantId, tool, templateId, e.getMessage(), e);
                    }
                } else if ("TideAi".equalsIgnoreCase(tool)) {
                    // Tide 线索导入：key 约定为 "TideAi-{callStrategyId}"
                    if (StrUtil.isBlank(taskId)) {
                        log.warn("[push-list] tenantId={} tool='{}' callStrategyId(taskId) is blank, skip this group. key={}, size={}",
                                tenantId, tool, key, groupItems.size());
                        continue;
                    }

                    String callStrategyId = taskId;
                    log.info("[push-list] tenantId={} tool='{}' import clue to TIDE, callStrategyId={}, size={}",
                            tenantId, tool, callStrategyId, groupItems.size());

                    try {
                        // 线索导入：bizData(JSON) = [{"taskName": "...","callStrategyId": "...","phone": "..."}]
                        TideTemplateSmsClient.importClueForPushItems(groupItems, null, callStrategyId);
                        // 同样：调用成功视为全组成功
                        for (PushItem it : groupItems) {
                            String sk = buildLogKey(it);
                            if (successKeySet.add(sk)) {
                                successItems.add(it);
                            }
                        }
                    } catch (Exception e) {
                        log.error("[push-list] tenantId={} tool='{}' import clue to TIDE error. callStrategyId={}, err={}",
                                tenantId, tool, callStrategyId, e.getMessage(), e);
                    }
                } else {
                    // 其他工具：整组一次性发送
                    log.info("[push-list] tenantId={} tool='{}' send as one batch, taskId(channel)='{}', size={}",
                            tenantId, tool, taskId, groupItems.size());
                    boolean sentOk = sendToInterfaceB(key, tool, groupItems, taskId);
                    if (sentOk) {
//                        for (PushItem it : groupItems) {
//                            String sk = buildLogKey(it);
//                            if (successKeySet.add(sk)) {
//                                successItems.add(it);
//                            }
//                        }
                    } else {
                        log.warn("[push-list] tenantId={} tool='{}' send failed. skip logging for this group.",
                                tenantId, tool);
                    }
                }
            } // end tool-group loop for this tenant

            // 2.3) 该 tenantId 下所有营销工具处理完成后，再统一落库（去重后）
            if (!successItems.isEmpty()) {
                log.info("[push-list] tenantId={} all tools done, persist logs, uniqueSuccess={}",
                        tenantId, successItems.size());
                persistGroupLogs(successItems);
            } else {
                log.info("[push-list] tenantId={} all tools done, no success items to persist.", tenantId);
            }
        }
    }

    /**
     * 构造用于去重的 key：
     * 同一 tenantId + listDate + listLevel + phoneE164 认为是同一条名单
     */
    private String buildLogKey(PushItem it) {
        return (it.getTenantId() == null ? "null" : String.valueOf(it.getTenantId()))
                + "|" + StrUtil.blankToDefault(it.getPhoneE164(), "")
                + "|" + StrUtil.blankToDefault(it.getListDate(), "")
                + "|" + StrUtil.blankToDefault(it.getListLevel(), "");
    }

    /**
     * 将某个租户内、所有成功发送的记录落库（逐条 upsert）。
     *
     * @param groupItems 该租户内需要落库的记录（已经在调用方按 buildLogKey 去重）
     * @return 实际成功落库的条数
     */
    private int persistGroupLogs(List<PushItem> groupItems) {
        int logged = 0;
        for (PushItem it : groupItems) {
            MarketingListLogDO vo = new MarketingListLogDO();
            vo.setPhoneE164(it.getPhoneE164());
            vo.setPhoneMd5(it.getPhoneMd5());
            vo.setScore(it.getScore());
            vo.setPhoneTags(it.getPhoneTags());
            vo.setRechargeTenantId(it.getTenantId());
            vo.setListLevel(it.getListLevel());
            vo.setListDate(it.getListDate());
            // 不设置 responseTime；固定 0
            vo.setResponseState(0);
            vo.setMarketingTools(it.getMarketingTools());

            // 若 DO 已支持 channel 字段且希望落库，可在此解除注释：
            // vo.setChannel(it.getChannel());

            try {
                marketingListLogService.saveOrUpdateIgnoreResponse(vo);
                logged++;
            } catch (Exception e) {
                log.error("[push-list] upsert log fail. phone={}, tenant={}, listDate={}, err={}",
                        it.getPhoneE164(), it.getTenantId(), it.getListDate(), e.getMessage(), e);
            }
        }
        log.info("[push-list] logged={} (group size={})", logged, groupItems.size());
        return logged;
    }

    /**
     * 94外呼：成功(code==200)返回 true，否则 false；成功时同时记录失败明细表
     */
    private boolean sendToInterface94(String marketingTool, List<PushItem> groupItems, String taskId) {
        log.info("[push-list] sending to Interface 94 with {} items", groupItems.size());

        final String requestId = UUID.randomUUID().toString();
        final String failReturn = "0"; // 返回 data 明细

        // 任务初次落表
        ActivityTaskDO task = new ActivityTaskDO();
        task.setRequestId(requestId);
        task.setImportNum(groupItems.size());
        task.setMarketingTool(marketingTool);
        activityTaskMapper.insert(task);

        String responseBody;
        try {
            responseBody = interfaceAClient.sendToInterfaceA(requestId, failReturn, groupItems, taskId);
        } catch (Exception e) {
            log.error("[push-list] 94waihu call error. requestId={}, err={}", requestId, e.getMessage(), e);
            task.setCode("500");
            task.setMessage("调用外呼接口异常：" + e.getMessage());
            activityTaskMapper.updateById(task);
            return false;
        }

        if (StrUtil.isBlank(responseBody)) {
            log.error("[push-list] 94waihu empty response. requestId={}", requestId);
            task.setCode("500");
            task.setMessage("空响应");
            activityTaskMapper.updateById(task);
            return false;
        }

        cn.hutool.json.JSONObject obj = JSONUtil.parseObj(responseBody);
        Integer code = obj.getInt("code");
        String message = obj.getStr("message");
        String respBatchId = obj.getStr("batchId");
        Integer respImportNum = obj.getInt("importNum");
        if (code == null || code != 200) {
            task.setCode(String.valueOf(code));
            task.setMessage("接口返回失败，code=" + code + ", message=" + message);
            activityTaskMapper.updateById(task);
            return false;
        }

        if (StrUtil.isNotBlank(respBatchId)) {
            task.setBatchId(respBatchId);
        }
        if (respImportNum != null) {
            task.setImportNum(respImportNum);
        }

        // 失败明细 data（仅 failReturn=0 返回）
        if (obj.containsKey("data") && obj.get("data") instanceof cn.hutool.json.JSONArray) {
            cn.hutool.json.JSONArray data = obj.getJSONArray("data");
            for (int i = 0; i < data.size(); i++) {
                cn.hutool.json.JSONObject it = data.getJSONObject(i);
                ActivityFailedNumbersDO failed = ActivityFailedNumbersDO.builder()
                        .taskId(task.getId())
                        .number(it.getStr("number"))
                        .indexs(it.getInt("index"))
                        .numberMd5(it.getStr("numberMD5"))
                        .tag(it.getStr("tag"))
                        .errorMessage(extractErrorMessage(it))
                        .build();
                failed.setCreateTime(LocalDateTime.now());
                activityFailedNumbersMapper.insert(failed);
            }
            log.info("[push-list] 94waihu requestId={} failedCount={}", requestId, data.size());
        }

        task.setCode("200");
        task.setMessage(StrUtil.blankToDefault(message, "数据推送成功"));
        activityTaskMapper.updateById(task);
        return true;
    }

    /**
     * 其他外呼（示例）：根据 tool 名称区分，成功才返回 true（此处保留对接点）
     */
    private boolean sendToInterfaceB(String marketingTool, String tool,
                                     List<PushItem> groupItems, String taskId) {
        log.info("[push-list] sending to Interface '{}' with {} items, taskId(channel)={}",
                tool, groupItems.size(), taskId);
        // TODO: 按照对应平台签名/请求/解析方式实现；返回 true 表示该组成功。
        // 这里先返回 false，避免误落库。
        return false;
    }

    /**
     * 提取错误信息（除 number/index/numberMD5/tag 外的键值；若 number 不是纯数字也保留）
     */
    private String extractErrorMessage(cn.hutool.json.JSONObject it) {
        Set<String> skip = new HashSet<>(Arrays.asList("number", "index", "numberMD5", "tag"));
        List<String> parts = new ArrayList<>();

        String numberVal = it.getStr("number");
        if (StrUtil.isNotBlank(numberVal) && !numberVal.matches("^\\d+$")) {
            parts.add("number:" + numberVal);
        }

        for (String key : it.keySet()) {
            if (skip.contains(key)) continue;
            Object v = it.get(key);
            if (v == null) continue;
            String val = String.valueOf(v);
            if (StrUtil.isBlank(val)) continue;
            parts.add(key + ":" + val);
        }
        return parts.isEmpty() ? null : String.join("; ", parts);
    }

    private static String toDay8(String in) {
        String s = in == null ? "" : in.trim();
        if (s.matches("^\\d{8}$")) return s;
        if (s.matches("^\\d{2,4}-\\d{1,2}-\\d{1,2}$")) return s.replace("-", "");
        return LocalDate.now().minusDays(1).format(DateTimeFormatter.ofPattern("yyyyMMdd"));
    }

    private static String toYmd(String day8) {
        return day8.substring(0, 4) + "-" + day8.substring(4, 6) + "-" + day8.substring(6, 8);
    }

    /**
     * 简易查询串解析：k1=v1&k2=v2...
     */
    static class Query {
        static Map<String, String> parse(String s) {
            HashMap<String, String> m = new HashMap<>();
            if (s == null || s.trim().isEmpty()) return m;
            for (String kv : s.split("&")) {
                if (kv.isEmpty()) continue;
                int i = kv.indexOf('=');
                if (i < 0) m.put(kv.trim(), "");
                else m.put(kv.substring(0, i).trim(), kv.substring(i + 1).trim());
            }
            return m;
        }
    }
}
